package com.shanks.framework.mq.processor;

import com.shanks.framework.mq.common.util.LogUtil;
import com.shanks.framework.mq.common.util.MQContext;
import com.shanks.framework.mq.domain.MQMessage;
import com.shanks.framework.mq.enums.MQStatusEnum;
import com.shanks.framework.mq.exception.MQException;
import com.shanks.framework.mq.helper.MqHelper;
import com.shanks.framework.mq.producer.BaseMQProducer;
import com.shanks.framework.mq.service.MessageConsumeService;
import com.shanks.framework.mq.common.util.DateUtil;
import jakarta.annotation.Resource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.transaction.support.TransactionSynchronizationAdapter;
import org.springframework.transaction.support.TransactionSynchronizationManager;

import java.util.Objects;

public abstract class AbstractMQProcessor implements MQProcessor {

    private final Logger logger = LoggerFactory.getLogger(getClass());

    private BaseMQProducer mqProducer;

    @Resource
    private MessageConsumeService messageConsumeService;

    @Resource
    private MqHelper mqHelper;

    public void setMqProducer(BaseMQProducer mqProducer) {
        this.mqProducer = mqProducer;
    }

    @Override
    public Boolean handleMessage(MQMessage mqMessage) throws Exception {
        try {
            // 验证消息签名 (假设这是由其他地方实现)

            // 检查消息是否已经被消费过
            if (messageConsumeService.messageHasConsume(mqMessage)) {
                logger.info("消息已消费：msgId={}, topic={}, tag={}, key={}",
                        mqMessage.getMsgId(), mqMessage.getTopic(), mqMessage.getTags(), mqMessage.getKey());
                return true;
            }

            // 具体消费消息并获取消费状态
            MQStatusEnum mqStatus = consumeMessage(mqMessage);

            // 添加消费记录
            messageConsumeService.addConsumeRecord(mqMessage, mqProducer.getNamesrvAddr(), mqHelper.getInstanceName());

            // 根据消费状态处理后续逻辑
            if (mqStatus == MQStatusEnum.SUCCESS) {
                // 消费成功后删除可能存在的异常记录
                messageConsumeService.deleteConsumeException(mqMessage);
                // 发送ACK确认消息已被成功处理
                sendAck(mqMessage, Boolean.TRUE);
                logger.info("消息消费成功：msgId={}, topic={}, tag={}, key={}",
                        mqMessage.getMsgId(), mqMessage.getTopic(), mqMessage.getTags(), mqMessage.getKey());
                return true;
            } else {
                // 处理消费失败的情况
                handleConsumeFailure(mqMessage);
                return false;
            }
        } catch (Exception e) {
            // 记录错误日志并添加消费异常记录
            logAndRecordError(mqMessage, e);
            throw e;
        } finally {
            // 清除线程上下文中的MQContext信息
            MQContext.remove();
        }
    }

    /**
     * 处理消费失败的情况。
     *
     * @param mqMessage 消息对象
     */
    private void handleConsumeFailure(MQMessage mqMessage) throws Exception {
        MQContext.MQErrorInfo errorInfo = MQContext.getErroInfo();

        if (Objects.isNull(errorInfo)) {
            throw new MQException(String.format("消息key : %s ,消息需要重发, 但未设置重发原因, 请在上下文线程MQContext中设置原因", mqMessage.getKey()));
        } else {
            // 抛出上下文设置的异常
            if (errorInfo.getException() != null) {
                throw errorInfo.getException();
            }
        }
    }

    /**
     * 记录错误日志并添加消费异常记录。
     *
     * @param mqMessage 消息对象
     * @param e         异常对象
     */
    private void logAndRecordError(MQMessage mqMessage, Exception e) throws Exception {
        logger.error("[{}] 处理消息失败.{}",
                mqMessage.getTopic() + "." + mqMessage.getTags(),
                LogUtil.formatAlert(
                        "MQ_WARN:CONSUME_FAILED",
                        "topicTag", mqMessage.getTopic() + "." + mqMessage.getTags(),
                        "msgKey", mqMessage.getKey(),
                        "exceptionMsg", e.getMessage()
                ), e);

        messageConsumeService.addConsumeExceptionRecord(mqMessage, mqHelper.getInstanceName());
    }

    protected void sendAck(final MQMessage message, final Boolean result) {
        // 存在事务时等待当前事务提交后再发送
        if (TransactionSynchronizationManager.isActualTransactionActive()) {
            TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
                @Override
                public void afterCommit() {
                    doSendAckInTranc(message, result);
                }
            });
        } else {
            doSendAck(message, result);
        }
    }

    private void doSendAckInTranc(final MQMessage message, final Boolean result) {
        try {
            mqProducer.sendAck(message.getTopic(), message.getTags(), message.getKey(), result, mqHelper.getInstanceName());
            logger.info("发送ACK消息完成(事务), 当前时间={}, msgId={}, topic={}, tag={}, key={} ", DateUtil.getCurrentTime(), message.getMsgId(), message.getTopic(), message
                    .getTags(), message.getKey());
        } catch (Exception e) {
            logger.error("发送ACK消息失败(事务), 当前时间={}, msgId={}, topic={}, tag={}, key={} ", DateUtil.getCurrentTime(), message.getMsgId(), message.getTopic(), message
                    .getTags(), message.getKey(), e);
        }
    }

    private void doSendAck(final MQMessage message, final Boolean result) {
        try {
            this.mqProducer.sendAck(message.getTopic(), message.getTags(), message.getKey(), result, mqHelper.getInstanceName());
            this.logger.info("发送ACK消息完成(非事务), 当前时间={}, msgId={}, topic={}, tag={}, key={} ", DateUtil.getCurrentTime(), message.getMsgId(), message.getTopic(), message
                    .getTags(), message.getKey());
        } catch (Exception e) {
            this.logger.error("发送ACK消息失败(非事务), 当前时间={}, msgId={}, topic={}, tag={}, key={} ", DateUtil.getCurrentTime(), message.getMsgId(), message.getTopic(), message
                    .getTags(), message.getKey(), e);
        }
    }


    /**
     * 具体消费方法
     *
     * @param message
     * @return
     * @throws Exception
     */
    protected abstract MQStatusEnum consumeMessage(MQMessage message) throws Exception;

    /**
     * 消息描述
     *
     * @param message
     * @return
     */
    protected abstract String getMessageDesc(MQMessage message);
}
